package com.un.ebs.sys.service.impl;

import com.un.ebs.sys.base.QueueTaskListener;
import com.un.ebs.sys.dto.DelayTaskDto;
import com.un.ebs.sys.service.DelayedQueueService;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.*;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

@Service
public class DelayedQueueServiceImpl implements DelayedQueueService {

    @Autowired
    RedissonClient redissonClient;

    static Map<Class, QueueTaskListener> pool = new HashMap<>();
    static List<DelayTaskDto> sendPool = Collections.synchronizedList(new ArrayList<>());
    static List<DelayTaskDto> revPool = Collections.synchronizedList(new ArrayList<>());
    CountDownLatch lock = new CountDownLatch(1);
    private boolean init = false;

    @Override
    public <T> void sendQueue(T t, long delay, TimeUnit timeUnit) {
        if (!init) {
            this.listenQueue();
        }
        DelayTaskDto<T> data = new DelayTaskDto<T>();
        data.setDto(t);
        data.setZClass(t.getClass());
        data.setSec(delay);
        data.setUnit(timeUnit);
        sendPool.add(data);
    }

    @Override
    public <T> void setQueue(Class zClass, QueueTaskListener listener) {
        pool.put(zClass, listener);
    }

    @Override
    public void listenQueue() {
        final String queueName = "delayQueue3";
        if (init) return;

        // rev thread
        new Thread(new Runnable() {
            @Override
            public void run() {
                RBlockingQueue<DelayTaskDto> blockingFairQueue = redissonClient.getBlockingQueue(queueName);
                lock.countDown();
                while (true) {
                    try {
                        DelayTaskDto t = blockingFairQueue.take();
                        revPool.add(t);
                        //System.out.println("get message");
                    } catch (Exception e) {
                    }
                }
            }
        }).start();
        // send thread
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    if (init && sendPool.size() != 0) {
                        try {
                            Thread.sleep(500);
                            DelayTaskDto dto = sendPool.get(0);
                            sendPool.remove(0);
                            RBlockingQueue<DelayTaskDto> blockingFairQueue = redissonClient.getBlockingQueue(queueName);
                            RDelayedQueue<DelayTaskDto> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
                            delayedQueue.offer(dto, dto.getSec(), dto.getUnit());
                            //delayedQueue.destroy();
                            //System.out.println("send message");
                        } catch (Exception e) {

                        }
                    }
                }
            }
        }).start();
        // handler thread
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    if (init && revPool.size() != 0) {
                        DelayTaskDto dto = revPool.get(0);
                        revPool.remove(0);
                        if (pool.containsKey(dto.getZClass())) {
                            QueueTaskListener listener = pool.get(dto.getZClass());
                            listener.invoke(dto.getDto());
                            //System.out.println("handler message");
                        }
                    }
                }
            }
        }).start();

        try {
            lock.await();
            this.init = true;
        } catch (Exception e) {
        }

    }

}
